package com.gitee.xhs.window.windowOperator;

import com.gitee.xhs.OperatorSourceFactory;
import com.gitee.xhs.customFunction.OperatorKeyBy;
import com.gitee.xhs.customWaterMark.CustomWaterStrategy;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @author BigHead
 * @date 2022/7/25 10:10
 * @description simpleWindowMain
 */
public class SimpleWindowMain {

    private final StreamExecutionEnvironment streamExecutionEnvironment;

    public SimpleWindowMain(StreamExecutionEnvironment streamExecutionEnvironment) {
        this.streamExecutionEnvironment = streamExecutionEnvironment;
    }

    public void windowLogic() {

        KeyedStream<Transaction, Long> windowKeyedTransactionStream = streamExecutionEnvironment
                .addSource(
                        OperatorSourceFactory.unBoundLoopOneToTenOneSecondSkipSource()
                )
                .assignTimestampsAndWatermarks(
                        new CustomWaterStrategy()
                )
                .keyBy(
                        new OperatorKeyBy()
                );

        // TumblingWindow
        tumblingWindowLogic(windowKeyedTransactionStream);
        // SlidingWindow
        slidingWindowLogic(windowKeyedTransactionStream);
        // EventTimeSessionWindow
        sessionWindowLogic(windowKeyedTransactionStream);
        // GlobalWindow 需要通过Trigger实现
    }

    private void tumblingWindowLogic(KeyedStream<Transaction, Long> windowKeyedTransactionStream) {
        windowKeyedTransactionStream
                .window(
                        TumblingEventTimeWindows.of(Time.minutes(1))
                )
                .aggregate(
                        new DefaultAggregateWindowFunction()
                )
                .print("TumblingWindow 效果 0,1,2,3,4,5 总计5条数据 6,7,8,9 总计6条数据");
    }

    private void slidingWindowLogic(KeyedStream<Transaction, Long> windowKeyedTransactionStream) {
        windowKeyedTransactionStream
                .window(
                        SlidingEventTimeWindows.of(Time.minutes(1), Time.seconds(30))
                )
                .aggregate(
                        new DefaultAggregateWindowFunction()
                )
                .print("SlidingEventWindow 效果 0,1,2,3,4,5 总计5条数据 6,7,8,9 总计6条数据");
    }

    private void sessionWindowLogic(KeyedStream<Transaction, Long> windowKeyedTransactionStream) {
        windowKeyedTransactionStream
                .window(EventTimeSessionWindows.withGap(Time.seconds(19)))
                .aggregate(
                        new DefaultAggregateWindowFunction()
                )
                .print("EventTimeSessionWindows 效果 0,1,2,3,4,5 5条数据 6,7,8,9 无数据");
    }

}
